#include<stdio.h>
#include<pthread.h>
#include<stdlib.h>
#include<unistd.h>
#include"ThreadPool.h"
#include"DoubleLinkList.h"
#include"LinkQueue.h"
#include"StdThread.h"
#include"pthread.h"

#define true 1
#define false 0
#define Sleep_Time 10
    
typedef void *(*function)(void *);
struct ThreadTask 
{
    function func;    //指针
    void *arg;      //任务
};
typedef struct ThreadTask task;

task *CreateTask(function func,void *arg)
{
    task *t=(task *)malloc(sizeof(task));
    if(t==NULL)
    {
        printf("CreateTask malloc error!\n");
        return NULL;
    }
    t->func=func;
    t->arg=arg;
}

void FreeTask(task *t)
{
    free(t);
}

struct ThreadPool   //线程池
{
    DLlist threads;         //线程们
    LQueue task_queue;      //队列组成的任务(先来的先处理)

    int max_thrd_num;
    int min_thrd_num;
    int max_queue_size;
    int busy_thrd_num;
    int waitkillthread;

    pthread_mutex_t pool_mutex;
    pthread_mutex_t busy_num_mutex;
    
    pthread_cond_t queue_not_empty;   //条件变量,判断队列中是否为空
    pthread_cond_t queue_not_full;    ////条件变量,判断队列中是否满了

    int shutdown;    //关掉线程池的开关

    thread *admin_thread;    //管理线程多少的线程,适当减少或增加
};
typedef struct ThreadPool ThreadP;

void *thread_work(void *arg)
{
    ThreadP *p=(ThreadP *)arg;
    while(1)
    {
        pthread_mutex_lock(&p->pool_mutex);   //上锁是为了防止两个线程同时操作队列
        while((IsQEmpty(&p->task_queue))==true && p->shutdown!=true)    //如果队列里没有任务就一直等待
        {
            pthread_cond_wait(&p->queue_not_empty,&p->pool_mutex);

            if(p->waitkillthread > 0)
            //kill
            if(p->busy_thrd_num*2 < GetListLen(&p->threads) && GetListLen(&p->threads) > p->min_thrd_num)
            {
                struct Node *TravelPoint=p->threads.head;
                while(TravelPoint!=NULL)
                {
                    thread *t=(thread *)TravelPoint->data;
                    if(GetThreadId(t)==pthread_self())
                    {
                        RemoveByElement(&p->threads,t);
                        break;
                    }
                    TravelPoint=TravelPoint->next;
                }
                pthread_mutex_unlock(&p->pool_mutex);
                pthread_exit(NULL);
            }
        }

        if(p->shutdown==true)    //关闭线程池
        {
            pthread_mutex_unlock(&p->pool_mutex);
            pthread_exit(NULL);
        }
        task *mission=(task*)(*(QPop(&p->task_queue)));   //把任务从队列中取出来
        pthread_cond_broadcast(&p->queue_not_full);   //取出来一个就告诉所有线程队列没满
        pthread_mutex_unlock(&p->pool_mutex);    //操作完队列之后就可以解除上锁了

        pthread_mutex_lock(&p->busy_num_mutex);
        p->busy_thrd_num++;
        pthread_mutex_unlock(&p->busy_num_mutex);   

        mission->func(mission->arg);     //执行任务  

        pthread_mutex_lock(&p->busy_num_mutex);
        p->busy_thrd_num--;
        pthread_mutex_unlock(&p->busy_num_mutex);     
    }
}

void *thread_manager(void *arg)
{
    ThreadP *p=(ThreadP *)arg;       //获取到整个线程池
    while(p->shutdown!=true)      // 线程池没关就继续操作
    {
       sleep(Sleep_Time);
       pthread_mutex_lock(&p->pool_mutex);     //操作线程池都要上锁

        //create
       if(GetLQueueLen(&p->task_queue) > p->busy_thrd_num && GetListLen(&p->threads) > p->max_thrd_num)
       {
            int addnum=(GetLQueueLen(&p->task_queue) - p->busy_thrd_num)/2;
            for(int i=0;i<addnum;i++)
            {
                thread *t=InitThread(thread_work,p);
                InsertTail(&p->threads,t);                
            }
       }

        int busy_thrd=p->busy_thrd_num;
        int threadLen=GetListLen(&p->threads);
        p->waitkillthread=((threadLen - busy_thrd)/2) > (threadLen - p->min_thrd_num) ? (threadLen - p->min_thrd_num) : ((threadLen - busy_thrd)/2);
        int minusNum=p->waitkillthread;
        pthread_mutex_unlock(&p->pool_mutex);
        //distory
        if(p->busy_thrd_num*2 < GetListLen(&p->threads) && GetListLen(&p->threads) > p->min_thrd_num)
        {
            for(int i=0;i<minusNum;i++)
            {
                pthread_cond_broadcast(&p->queue_not_empty);
            }
        }
    }
    pthread_exit(NULL);
}

ThreadP *InitThreadPool(int max_thrd_num, int min_thrd_num, int max_queue_size)
{
    ThreadP *p=(ThreadP *)malloc(sizeof(ThreadP));
    if(p==NULL)
    {
        printf("InitThreadPool malloc error!\n");
        return NULL;
    }
    InitDLinkList(&p->threads);
    InitLQueue(&p->task_queue);

    p->max_thrd_num=max_thrd_num;
    p->min_thrd_num=min_thrd_num;
    p->max_queue_size=max_queue_size;
    p->waitkillthread=0;
    p->busy_thrd_num=0;

    pthread_mutex_init(&p->pool_mutex,NULL);
    pthread_mutex_init(&p->busy_num_mutex,NULL);
    pthread_cond_init(&p->queue_not_empty,NULL);
    pthread_cond_init(&p->queue_not_full,NULL);

    for(int i=0;i<max_thrd_num;i++)
    {
        thread *t=InitThread(thread_work,p);
        InsertTail(&p->threads,t);
    }

    p->admin_thread=InitThread(thread_manager,p);
    return p;
}

void ThreadP_AddTask(ThreadP *p, void *(*func)(void *), void *arg)
{
    pthread_mutex_lock(&p->pool_mutex);
    while(GetLQueueLen(&p->task_queue)==p->max_queue_size && p->shutdown!=true)
    {
        pthread_cond_wait(&p->queue_not_full,&p->pool_mutex);
    }

    if(p->shutdown==true)
    {
        pthread_mutex_unlock(&p->pool_mutex);
        return;
    }

    QPush(&p->task_queue,CreateTask(func,arg));    //将新的任务加到队列里
    pthread_cond_broadcast(&p->queue_not_empty);   //告诉所有线程队列不为空
    pthread_mutex_unlock(&p->pool_mutex);     
}

void DestoryThreadPool(ThreadP *p)
{
    if(p==NULL)
    {
        return;
    }

    p->shutdown==true;

    JoinThread(p->admin_thread);

    int len=GetListLen(&p->threads);    //回收空闲的线程
    for(int i=0;i<len;i++)
    {
        pthread_cond_broadcast(&p->queue_not_empty);
    }

    struct Node *TravelPoint=p->threads.head;    //回收还在忙的线程
    while(TravelPoint!=NULL)
    {
        thread *t=(thread *)TravelPoint->data;
        JoinThread(t);      //这个函数会一直等待线程忙完
        free(t);
        TravelPoint=TravelPoint->next;
    }
    FreeDLinklist(&p->threads);

    while(IsQEmpty(&p->task_queue)!=true)       //释放队列
    {
        void *task=*QPop(&p->task_queue);
        free(task);
    }
    FreeLQueue(&p->task_queue);   

    pthread_mutex_destroy(&p->pool_mutex);          //释放锁
    pthread_mutex_destroy(&p->busy_num_mutex);

    pthread_cond_destroy(&p->queue_not_empty);    //释放条件变量
    pthread_cond_destroy(&p->queue_not_full);

    free(p);      //释放线程池本身
}
